/**   
 * @Title: AvroUtil.java 
 * @Package com.org.source.avro 
 * @Description: TODO(用一句话描述该文件做什么) 
 * @author cssuger@163.com   
 * @date 2016年12月21日 上午10:19:58 
 * @version V1.0   
 */
package com.biconn.bigdata.common.util;

import java.io.ByteArrayOutputStream;
import java.io.File;
import java.io.IOException;
import java.io.OutputStream;
import java.util.Collection;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;

import org.apache.avro.Protocol;
import org.apache.avro.Schema;
import org.apache.avro.Schema.Field;
import org.apache.avro.file.DataFileReader;
import org.apache.avro.file.DataFileWriter;
import org.apache.avro.generic.GenericArray;
import org.apache.avro.generic.GenericData;
import org.apache.avro.generic.GenericData.Record;
import org.apache.avro.generic.GenericDatumReader;
import org.apache.avro.generic.GenericDatumWriter;
import org.apache.avro.generic.GenericRecord;
import org.apache.avro.io.DatumReader;
import org.apache.avro.io.DatumWriter;
import org.apache.avro.io.Decoder;
import org.apache.avro.io.DecoderFactory;
import org.apache.avro.io.Encoder;
import org.apache.avro.io.EncoderFactory;
import org.apache.avro.specific.SpecificDatumReader;
import org.apache.commons.io.IOUtils;

import com.alibaba.fastjson.JSONObject;
import com.google.common.collect.Collections2;
import com.google.common.collect.Lists;
import com.google.common.collect.Sets;
import com.google.common.collect.Sets.SetView;



/**
 * @ClassName: AvroUtil
 * @Description: 可否将数据交换的利用avro进行统一的数据交换
 * @author cssuger@163.com
 * @date 2016年12月21日 上午10:19:58
 */
public class AvroUtil {

	private ResourceLoad resourceLioad;
	
	

	
	public AvroUtil() {

	}
	
	public AvroUtil(ResourceLoad resourceLioad) {
        this.resourceLioad = resourceLioad;
	}

	

	/**
	 * 
	 * @Title: parseProtocol
	 * @Description: 解析协议
	 * @param @param file
	 * @param @return 设定文件
	 * @return Protocol 返回类型
	 * @throws
	 */
	public Protocol parseProtocol() {
		return resourceLioad.loadProtocol();
	}

	/**
	 * 
	 * @Title: jsonToGenericRecord   
	 * @Description: json 反转为 GenericRecord  
	 * @param json
	 * @param schema
	 * @return      
	 * GenericRecord      
	 * @throws
	 */
	public GenericRecord jsonToGenericRecord(String json ,Schema schema) {
		DatumReader<GenericRecord> datumReader = new GenericDatumReader<GenericRecord>(schema);
		Decoder decoder;
		try {
			decoder = DecoderFactory.get().jsonDecoder(schema, json);
			return  datumReader.read(null, decoder);
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		return null;
		
	}
	
	

	/**
	 * 
	 * @Title: getAllSchemas
	 * @Description: 获取所有的type数据结构
	 * @param @return 设定文件
	 * @return List<Schema> 返回类型
	 * @throws
	 */
	public List<Schema> getAllSchemas() {
		Protocol protocol = this.parseProtocol();
		if (null == protocol) {
			throw new NullPointerException("protocol is nou null");
		}
		Collection<Schema> schemas = protocol.getTypes();
		return Lists.newArrayList(schemas);
	}
	
	/**
	 * 
	 * @Title: put   往结构里面增加数据
	 * @Description: TODO(这里用一句话描述这个方法的作用)   
	 * @param: protocol
	 * @param: name
	 * @param: values      
	 * @return:void      
	 * @throws
	 */
	public String put(Schema schema,String [] values) {
		GenericRecord requestData = new GenericData.Record(schema);
		for(int i =0;i<values.length;i++) {
			requestData.put(i, StringTools.mkString(values[i]));
		}
		return requestData.toString();
	}
	
	/**
	 * 
	 * @Title: copy   
	 * @Description: 两   Schema  互相拷贝
	 * @param sorceschema 源Schema
	 * @param disschema：目标Schema
	 * @param sorceGenericRecord 源Schema对应的值
	 * @return      
	 * GenericRecord      
	 * @throws
	 */
	public GenericRecord copy(Schema sorceschema,Schema disschema,GenericRecord sorceGenericRecord ) {
		GenericRecord requestData = new GenericData.Record(disschema);
		//也许存在两个schema里面的属性不一样的话
		List<Field> sourceFields =  sorceschema.getFields();
		List<Field> disFields =  disschema.getFields();
		//找到里面共有的属性
		SetView<Field> setView = Sets.intersection(Sets.newHashSet(sourceFields), Sets.newHashSet(disFields));
		setView.forEach(field ->{
			requestData.put(field.name(), sorceGenericRecord.get(field.name()));
		});
		//也许还会余下一些没有被赋值的字段，看会不会报错，报错了在在给赋值为"",他妈的
		return requestData;
	}
	
	
	/**
	 * 
	 * @Title: copy   
	 * @Description: 将json的值拷贝到sorceGenericRecord对象里面去，考虑到这连个结构会存在字段的数量不对等的情况
	 * @param jsonObject JSONObject
	 * @param disschema：目标Schema
	 * @param sorceGenericRecord 源Schema对应的值
	 * @return      
	 * GenericRecord      
	 * @throws
	 */
	public GenericRecord copy(JSONObject jsonObject,Schema disschema,GenericRecord sorceGenericRecord ) {
		GenericRecord requestData = new GenericData.Record(disschema);
		//也许存在两个schema里面的属性不一样的话
		Set<String> keys = jsonObject.keySet();
		List<Field> disFields =  disschema.getFields();
		Set<String> set = this.getFieldName(disFields);
		//找到里面共有的属性
		SetView<String> setView = Sets.intersection(keys, set);
		setView.forEach(name ->{
			requestData.put(name, jsonObject.getString(name));
		});
		
		//也许还会余下一些没有被赋值的字段，看会不会报错，报错了在在给赋值为"",他妈的
		return requestData;
	}

	
	
	/**
	 * 
	 * @Title: copy   
	 * @Description: 将json的值拷贝到sorceGenericRecord对象里面去，这种场景直接就是字段名称都不一样，这个时候需要进行源和目标字段进行对应
	 * @param jsonObject JSONObject
	 * @param disschema：目标Schema
	 * @param sorceGenericRecord 源Schema对应的值
	 * @param mapper 源和目标的对应关系，key为源，value为目标字段
	 * @return      
	 * GenericRecord      
	 * @throws
	 */
	public GenericRecord copy(JSONObject jsonObject,Schema disschema,GenericRecord sorceGenericRecord,Map<String,String> mapper ) {
		GenericRecord requestData = new GenericData.Record(disschema);
		//也许存在两个schema里面的属性不一样的话
		Set<String> keys = jsonObject.keySet();
		//List<Field> disFields =  disschema.getFields();
		Set<String> set = Sets.newHashSet(mapper.keySet());
		//找到里面共有的属性
		SetView<String> setView = Sets.intersection(keys, set);
		setView.forEach(name ->{
			String value_key = mapper.get(name);
			String value = jsonObject.getString(name);
			requestData.put(value_key, jsonObject.getString(value));
		});
		
		//也许还会余下一些没有被赋值的字段，看会不会报错，报错了在在给赋值为"",他妈的
		return requestData;
	}

	
	
	private Set<String> getFieldName(List<Field> disFields ){
		Set<String> set = Sets.newHashSet();
	   for(Field field : disFields) {
		   set.add(field.name());
	   }
	   return set;
	}
	
	public void add(Schema schema,Object object,int capacity) {
		GenericArray<Object> genericarray =	new GenericData.Array<Object>(capacity, schema);
		genericarray.add(object);
	}
	
	 

	/**
	 * 
	 * @Title: getSchemasById
	 * @Description: "types": [ {"name":"LogEntity", "type":"record", "fields":[
	 *               {"name":"body", "type":["string", "null"]},
	 *               {"name":"timestamp", "type":["string", "null"]},
	 *               {"name":"header", "type":["string", "null"]}
	 * 
	 *               ]},
	 * 
	 *               ],
	 * @param typeName 表示一个record的唯一标示,上面的LogEntity就是typeName
	 * @param @return 设定文件
	 * @return Schema 返回类型
	 * @throws
	 */
	public Schema getSchemasBytypename(String typeName) {
		Protocol protocol = this.parseProtocol();
		if (null == protocol) {
			throw new NullPointerException("protocol is nou null");
		}
		Schema entitySchema = protocol.getType(typeName);
		return entitySchema;
	}
	
	public Collection<Schema> findAllSchemas(){
		Protocol protocol = this.parseProtocol();
		if (null == protocol) {
			throw new NullPointerException("protocol is nou null");
		}
		return protocol.getTypes();
	}
	
	

	/**
	 * 
	 * @Title: getAllFieldBySchema
	 * @Description: 获取某个
	 * @param @param typeName
	 * @param @return 设定文件
	 * @return List<Field> 返回类型
	 * @throws
	 */
	public List<Field> getAllFieldBySchema(String typeName) {
		Schema entitySchema = this.getSchemasBytypename(typeName);
		return entitySchema.getFields();
	}

	/**
	 * 
	 * @Title: serializing 
	 * @Description: 这种的是序列化到文件里面 
	 * @param @param schema
	 * @param @param genericrecord
	 * @param @param file
	 * @param @throws IOException    设定文件 
	 * @return void    返回类型 
	 * @throws
	 */
	public void serializing(Schema schema, Record genericrecord, File file)
			throws IOException {

		DatumWriter<GenericRecord> userDatumWriter = new GenericDatumWriter<GenericRecord>();
		OutputStream outputStream = new ByteArrayOutputStream();
		Encoder encoder = EncoderFactory.get().jsonEncoder(schema, outputStream);
		userDatumWriter.write(genericrecord, encoder);
		DataFileWriter<GenericRecord> dataFileWriter = new DataFileWriter<GenericRecord>(
				userDatumWriter);
		DataFileWriter<GenericRecord> dfw = null;
		if (!file.exists()) {
			dfw = dataFileWriter.create(schema, file);
		} else {
			dfw = dataFileWriter.appendTo(file);
		}
		try {
			dfw.append(genericrecord);

		} catch (Exception e) {
			e.printStackTrace();
			// logger.error("write local file [" + filename + "] error", e);
		} finally {
			dfw.close();
			dataFileWriter.close();
		}

	}

	
	
//	public void deserializing(Schema schema, File file) throws IOException {
//		DatumReader<GenericRecord> datumReader = new GenericDatumReader<GenericRecord>(
//				schema);
//		DataFileReader<GenericRecord> dataFileReader = new DataFileReader<GenericRecord>(
//				file, datumReader);
//		GenericRecord user = null;
//		try{
//			while (dataFileReader.hasNext()) {
//				// Reuse user object by passing it to next(). This saves us from
//				// allocating and garbage collecting many objects for files with
//				// many items.
//				user = dataFileReader.next(user);
//				System.out.println(user);
//				
//			}
//		}finally{
//			dataFileReader.close();
//		}
//		
//	}

	public void writeToLocalFile(Schema schema, Record genericrecord)
			throws IOException {

	}

	public static void main(String[] args) throws IOException {
		String str = IOUtils.toString(new File("D:\\git\\big-data-support\\big-data-support\\big-data-support-jstorm-question-and-answer\\src\\main\\resources\\botu.avsc").toURI());
		AvroUtil util = new AvroUtil(new StringResourceLoad(str));
		Schema schema =	util.getSchemasBytypename("BusinessLog");
		//String json = "{\"id\": \"cxh\",\"operater\": \"man\",\"bus_name\": \"def\",\"befor_hander\": \"def\",\"key\": \"dgh\",\"result\": \"dgh\",\"reson\": \"dhjk\",\"datetime\": \"2018-8-13\"}";
		//GenericRecord genericRecord = util.jsonToGenericRecord(json, schema);
		//System.out.println(genericRecord.toString());
		String [] values = {new UIDGenerator().generate().toString(),"ADMIN","规则编辑","1111111111111111","","1","1",null};
		String json = util.put(schema, values);
		System.out.println("json:"+json);
		Schema schema1 =	util.getSchemasBytypename("BusinessLog");
		GenericRecord genericrecord =	util.jsonToGenericRecord(json, schema1);
		System.out.println(genericrecord);
		

	}
}
